package com.we.flink.userlabelplatform.tag;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Lists;
import com.we.flink.userlabelplatform.table.TableUserLoan;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;

public class TagUserLoan {
    public long uid;
    public int payoff_flag;
    public Integer first_loan_time;
    public Integer last_apply_time;
    public Integer last_user_loan_expire_time;

    public static Logger LOG = LoggerFactory.getLogger(TagUserLoan.class);

    public TagUserLoan() {}

    public TagUserLoan(
            long uid,
            int payOffFlag,
            Integer firstLendTime,
            Integer lastpplyTime,
            Integer lastUserLoanExpireTime) {
        this.uid = uid;
        this.payoff_flag = payOffFlag;
        this.first_loan_time = firstLendTime;
        this.last_apply_time = lastpplyTime;
        this.last_user_loan_expire_time = lastUserLoanExpireTime;
    }

    public static TagUserLoan coInit(long uid) {
        return new TagUserLoan(uid, 2, null, null, null);
    }

    public static TagUserLoan init(
            long uid,
            int payOffFlag,
            Integer firstLendTime,
            Integer lastpplyTime,
            Integer lastUserLoanExpireTime) {
        return new TagUserLoan(
                uid, payOffFlag, firstLendTime, lastpplyTime, lastUserLoanExpireTime);
    }

    public static TagUserLoan init(long uid) {
        return new TagUserLoan(uid, 2, Integer.MAX_VALUE, Integer.MIN_VALUE, Integer.MIN_VALUE);
    }

    public static SingleOutputStreamOperator<String> getFlatMapStream(
            StreamTableEnvironment tableEnv, StreamExecutionEnvironment env) {
        Table tableRes = tableEnv.sqlQuery(TableUserLoan.QUERYSQL);
        DataStream<Tuple2<Boolean, TableUserLoan>> inputStream =
                tableEnv.toRetractStream(tableRes, TableUserLoan.class);

        SingleOutputStreamOperator<String> out =
                inputStream
                        .filter(
                                new FilterFunction<Tuple2<Boolean, TableUserLoan>>() {
                                    @Override
                                    public boolean filter(Tuple2<Boolean, TableUserLoan> value)
                                            throws Exception {
                                        return value.f0;
                                    }
                                })
                        .uid("TagUserLoan-filter")
                        .keyBy(
                                new KeySelector<Tuple2<Boolean, TableUserLoan>, Long>() {
                                    @Override
                                    public Long getKey(Tuple2<Boolean, TableUserLoan> value)
                                            throws Exception {
                                        return value.f1.getUid();
                                    }
                                })
                        .flatMap(new UserLoanTagFlatMapFunction());
        return out;
    }

    private static class UserLoanTagFlatMapFunction
            extends RichFlatMapFunction<Tuple2<Boolean, TableUserLoan>, String> {

        MapState<Integer, Integer> loanSuccessMState;
        MapState<Integer, Integer> repaySuccessMState;
        MapState<Integer, Integer> applyTtimeMState;
        MapState<Integer, Integer> loanTimeMState;
        ValueState<TagUserLoan> userLoanTagVState;

        @Override
        public void open(Configuration parameters) throws Exception {
            applyTtimeMState =
                    getRuntimeContext()
                            .getMapState(
                                    new MapStateDescriptor<Integer, Integer>(
                                            "ApplyTtimeMState", Integer.class, Integer.class));
            loanTimeMState =
                    getRuntimeContext()
                            .getMapState(
                                    new MapStateDescriptor<Integer, Integer>(
                                            "LoanTimeMState", Integer.class, Integer.class));
            loanSuccessMState =
                    getRuntimeContext()
                            .getMapState(
                                    new MapStateDescriptor<Integer, Integer>(
                                            "LoanSuccessMState", Integer.class, Integer.class));
            repaySuccessMState =
                    getRuntimeContext()
                            .getMapState(
                                    new MapStateDescriptor<Integer, Integer>(
                                            "RepaySuccessMState", Integer.class, Integer.class));
            userLoanTagVState =
                    getRuntimeContext()
                            .getState(
                                    new ValueStateDescriptor<TagUserLoan>(
                                            "UserLoanTagVState", TagUserLoan.class));
        }

        @Override
        public void flatMap(Tuple2<Boolean, TableUserLoan> input, Collector<String> out)
                throws Exception {
            Integer firstLoanTime = Integer.MAX_VALUE;
            Integer lastApplyTime = Integer.MIN_VALUE;
            Integer lastExpireTime = Integer.MIN_VALUE;
            int payOffFlag = 2;

            int borrowTime = input.f1.getBorrow_time();
            int emitTime = input.f1.getEmit_time();
            int expire_time = input.f1.getExpire_time();
            int loanId = input.f1.getLoan_id();
            int status = input.f1.getStatus();
            int repayStatus = input.f1.getRepay_status();
            long uid = input.f1.getUid();

            TagUserLoan tmpLoanTag = userLoanTagVState.value();
            TagUserLoan curLoanTag = tmpLoanTag == null ? TagUserLoan.init(uid) : tmpLoanTag;
            try {
                /** update binlog process */
                /** 最近一次申请时间 */
                applyTtimeMState.put(loanId, borrowTime);
                /** 账龄:第一次放款成功时间 只要放款成功就加入LoanSuccessMapState */
                if (status == 5) {
                    loanTimeMState.put(loanId, emitTime);
                    loanSuccessMState.put(loanId, status);
                    lastExpireTime =
                            expire_time > curLoanTag.getLast_user_loan_expire_time()
                                    ? expire_time
                                    : curLoanTag.getLast_user_loan_expire_time();
                }
                /** 最后应还款日 */

                /** 放款+还款成功才能加入RepaySuccessMapState */
                if (repayStatus == 3 && status == 5) {
                    repaySuccessMState.put(loanId, repayStatus);
                }

                ArrayList<Integer> loanTimelists = Lists.newArrayList(loanTimeMState.values());
                ArrayList<Integer> applyTimeLists = Lists.newArrayList(applyTtimeMState.values());
                ArrayList<Integer> loanSuccessIdLists =
                        Lists.newArrayList(loanSuccessMState.keys());
                ArrayList<Integer> repaySuccessIdLists =
                        Lists.newArrayList(repaySuccessMState.keys());

                for (int time : loanTimelists) {
                    if (time < firstLoanTime) {
                        firstLoanTime = time;
                    }
                }

                for (int time : applyTimeLists) {
                    if (time > lastApplyTime) {
                        lastApplyTime = time;
                    }
                }

                int loanSuccSize = loanSuccessIdLists.size();
                int repaySuccSize = repaySuccessIdLists.size();
                if (loanSuccSize == 0) {
                    /** 未签约 */
                    payOffFlag = 2;
                } else {
                    if (loanSuccSize == repaySuccSize) {
                        /** 结清 */
                        payOffFlag = 1;
                    } else if (loanSuccSize > repaySuccSize) {
                        /** 放款成功 > 还款成功 未结清 */
                        payOffFlag = 0;
                    } else {
                        /** 放款成功 < 还款成功 出错 这个好像没法测试到 */
                        payOffFlag = 3;
                        Iterable<Integer> loanKeys = loanSuccessMState.keys();
                        LOG.error(loanKeys.toString());
                        Iterable<Integer> repayKeys = repaySuccessMState.keys();
                        LOG.error(repayKeys.toString());
                    }
                }
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("uid", uid);

                if (!curLoanTag.getFirst_loan_time().equals(firstLoanTime)) {
                    jsonObject.put(
                            "first_loan_time",
                            (firstLoanTime == Integer.MAX_VALUE ? null : firstLoanTime));
                    curLoanTag.first_loan_time = firstLoanTime;
                }

                if (curLoanTag.getPayoff_flag() != payOffFlag) {
                    jsonObject.put("payoff_flag", payOffFlag);
                    curLoanTag.payoff_flag = payOffFlag;
                }

                if (!curLoanTag.getLast_apply_time().equals(lastApplyTime)) {
                    jsonObject.put(
                            "last_apply_time",
                            (lastApplyTime == Integer.MIN_VALUE ? null : lastApplyTime));
                    curLoanTag.last_apply_time = lastApplyTime;
                }

                if (lastExpireTime > curLoanTag.getLast_user_loan_expire_time()) {
                    jsonObject.put("last_user_loan_expire_time", lastExpireTime);
                    curLoanTag.last_user_loan_expire_time = lastExpireTime;
                }

                if (jsonObject.size() > 1) {
                    out.collect(jsonObject.toString());
                    userLoanTagVState.update(curLoanTag);
                }

            } catch (Exception E) {
                LOG.error(E.getMessage());
                E.printStackTrace();
            }
        }
    }

    @Override
    public String toString() {
        return "TagUserLoan{"
                + "uid="
                + uid
                + ", payoff_flag="
                + payoff_flag
                + ", first_loan_time="
                + first_loan_time
                + ", last_apply_time="
                + last_apply_time
                + ", last_user_loan_expire_time="
                + last_user_loan_expire_time
                + '}';
    }

    public long getUid() {
        return uid;
    }

    public int getPayoff_flag() {
        return payoff_flag;
    }

    public Integer getFirst_loan_time() {
        return first_loan_time;
    }

    public Integer getLast_apply_time() {
        return last_apply_time;
    }

    public Integer getLast_user_loan_expire_time() {
        return last_user_loan_expire_time;
    }
}
